package com.yyx.jupiter.service.recommend.recall;

import com.alibaba.fastjson2.JSONObject;
import com.alibaba.nacos.common.utils.StringUtils;
import com.yyx.jupiter.conf.PropertyConfig;
import com.yyx.jupiter.entity.RecomReq;
import com.yyx.jupiter.entity.RecomVideo;
import com.yyx.jupiter.component.RedisClient;
import com.yyx.jupiter.entity.UpdateVideo;
import com.yyx.jupiter.utils.JsonUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.BeanWrapper;
import org.springframework.beans.BeanWrapperImpl;
import jakarta.annotation.Resource;
import org.springframework.stereotype.Service;

import java.beans.PropertyDescriptor;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;

@Slf4j
@Service
public class RecallManager {

    @Resource
    PropertyConfig propertyConfig;

    @Resource
    RedisClient redisClient;

    @Resource
    HotRecall hotRecall;

    @Resource
    NewVideoRecall newVideoRecall;

    private List<Recallable<RecomVideo>> getRecallStrategies(RecomReq recomReq) throws Exception {
        List<Recallable<RecomVideo>> recallStrategyList = new LinkedList<>();

        JSONObject strategyJson = propertyConfig.getStrategy().getJSONObject(recomReq.getStrategy());
        String recallPriority = strategyJson.getString("recallPriority");
        String[] recallNames = recallPriority.split(",");
        for (String recallName : recallNames) {
            switch (recallName) {
                case "hot":
                    recallStrategyList.add(hotRecall);
                    break;
                case "newVideo":
                    recallStrategyList.add(newVideoRecall);
                    break;
            }
        }
        return recallStrategyList;
    }

    private static ConcurrentHashMap<Integer, Map<String, RecomVideo>> concurrentRecall(
            ExecutorService executorService,
            List<Recallable<RecomVideo>> recallStrategies,
            RecomReq recomReq) throws Exception {
        ConcurrentHashMap<Integer, Map<String, RecomVideo>> recallVideoMap = new ConcurrentHashMap<>();
        CountDownLatch countDownLatch = new CountDownLatch(recallStrategies.size());
        for (int i = 0; i < recallStrategies.size(); i++) {
            int priorIdx = i;
            executorService.execute(() -> {
                try {
                    Map<String, RecomVideo> recomVideoMap = new LinkedHashMap<>();
                    List<RecomVideo> recomVideoList = recallStrategies.get(priorIdx).recall(recomReq);
                    for (RecomVideo recomVideo : recomVideoList) {
                        recomVideoMap.put(recomVideo.getVideoId(), recomVideo);
                    }
                    recallVideoMap.put(priorIdx, recomVideoMap);
                } catch (Exception e) {
                    log.error("concurrentRecall has error = ", e);
                    throw new RuntimeException(e);
                } finally {
                    countDownLatch.countDown();
                }
            });
        }

        countDownLatch.await();
        return recallVideoMap;
    }


    public List<RecomVideo> getRecallVideos(RecomReq recomReq, ExecutorService executorService) throws Exception {
        LinkedHashMap<String, RecomVideo> recomVideoMap = new LinkedHashMap<>();

        // 获取召回列表
        List<Recallable<RecomVideo>> recallerList = getRecallStrategies(recomReq);

        // 并行请求召回短视频
        ConcurrentHashMap<Integer, Map<String, RecomVideo>> recallVideoMap = concurrentRecall(executorService, recallerList, recomReq);

        // 召回短视频去重
        for (int i = recallVideoMap.size() - 1; i >= 0; i--) {
            recomVideoMap.putAll(recallVideoMap.get(i));
        }

        return new ArrayList<>(recomVideoMap.values());
    }

    public List<RecomVideo> fillRecomVideoData(List<RecomVideo> recomVideos) {
        LinkedHashMap<String, RecomVideo> recomVideoMap = new LinkedHashMap<>();
        for (RecomVideo recomVideo : recomVideos) {
            String itemKey = String.format("yyx:video:%s", recomVideo.getVideoId());
            recomVideoMap.put(itemKey, recomVideo);
        }

        List<String> recomVideoKeys = new ArrayList<>(recomVideoMap.keySet());
        Map<String, Map<String, Object>> objects = redisClient.multiHgetAll(recomVideoKeys);
        for (Map.Entry<String, Map<String, Object>> entry : objects.entrySet()) {
            if (entry.getValue() == null) {
                log.error("{}获取不到正排信息!", entry.getKey());
                continue;
            }
            RecomVideo recomVideo = JSONObject.parseObject(JsonUtils.toJsonString(entry.getValue()), RecomVideo.class);
            RecomVideo originalRecomVideo = recomVideoMap.get(entry.getKey());
            BeanUtils.copyProperties(recomVideo, originalRecomVideo, getNullPropertyNames(recomVideo));
        }
        return recomVideos;
    }

    public List<RecomVideo> filterRemovedVideos(List<RecomVideo> recomVideos) {
        String redisKey = "yyx:video_remove";
        String redisObj = redisClient.get(redisKey, String.class);
        if (redisObj == null) {
            return recomVideos;
        }
        List<String> removeVideos = Arrays.asList(redisObj.split(","));
        return recomVideos.stream().filter(recomVideo -> !removeVideos.contains(recomVideo.getVideoId())).toList();
    }

    private static String[] getNullPropertyNames(Object obj) {
        final BeanWrapper objWrapper = new BeanWrapperImpl(obj);
        PropertyDescriptor[] propertyDescriptors = objWrapper.getPropertyDescriptors();

        Set<String> emptyProps = new HashSet<>();
        for (PropertyDescriptor propertyDescriptor : propertyDescriptors) {
            Object propertyValue = objWrapper.getPropertyValue(propertyDescriptor.getName());
            if (propertyValue == null) {
                emptyProps.add(propertyDescriptor.getName());
            }
        }
        String[] result = new String[emptyProps.size()];
        return emptyProps.toArray(result);
    }

    public void updateNewVideoRecall(UpdateVideo updateVideo) throws Exception {
        String newRecallKey = "yyx:new_video_recall:" + updateVideo.getFirstCategory();
        String newVideoIds = redisClient.get(newRecallKey, String.class);
        if (StringUtils.contains(newVideoIds, updateVideo.getVideoId())) {
            return;
        }
        newVideoIds += "," + updateVideo.getVideoId();
        redisClient.setEx(newRecallKey, newVideoIds, 60L * 60 * 24 * 2);
    }
}
